package com.mango.ch00;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * job提交器实际是一个yarn集群的客户端， 它负责将我们mr程序需要的信息全部封装成一个配置文件
 * 然后连同我们的mr程序所在的jar包，一起提交给yarn去启动我们的mappmaster
 *
 * @author 游客
 */

// com.mango.mapreduce.JobClient
public class WordCountJob extends Configured implements Tool {
    //        static Path outPath = new Path("/WordCountJob/output");
    static Path localInput = new Path("D:/HadoopData/input/");
    static Path localOutput = new Path("D:/HadoopData/output/");
//    static Path inputPath = new Path("/MBA_Job/input");

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Configuration(), new WordCountJob(), args);
    }

    @Override
    public int run(String[] args) throws Exception {
        // 创建输入文件夹
        /*
         * try { HDFSAPI hdfs = new HDFSAPI(); hdfs.createDirectory(inputPath); //
         * 先删除已经有的输出文件夹 hdfs.delDirectory(outPath); // hdfs.orpOver(); } catch
         * (IOException e1) {
         * System.out.println("----------文件操作失败"); } catch (InterruptedException e) { //
         *  catch block e.printStackTrace(); } catch
         * (URISyntaxException e) {
         * e.printStackTrace(); }
         */
        Configuration conf = getConf();
        // 创建一个job提交对象
        Job job = Job.getInstance(conf);

        // 告知客户端提交器，程序启动使用的jar包（本程序中使用了main
        // 并且本class与业务逻辑的代码打包在一个jar包中，可以动态的获取到jar包所在的位置）
        job.setJarByClass(WordCountJob.class);

        // 告诉marappmaster 我们程序里要使用的业务实现类
        job.setMapperClass(WordCoutnMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // 告知mrappmaster我们这个程序map和reduce阶段输入输出的数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 告知marappmaster 我们要求要启动的reducetask的数量
        job.setNumReduceTasks(2);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 告知mrappmaster 我们程序要处理的数据来源所在的目录，以及我们要求的输出结果所在的目录
        FileInputFormat.setInputPaths(job, localInput);
        FileOutputFormat.setOutputPath(job, localOutput);
        Boolean res = job.waitForCompletion(true);// true 会在客户端打印一些信息
        System.out.println("Counter is " + job.getCounters().findCounter("Hello", "num").getValue());
        System.exit(res ? 0 : 1);
        return 0;
    }

}
